#pragma once
#include "../common/dispatcher.hpp"
#include "requestor.hpp"
#include "rpc_registry.hpp"
#include "rpc_caller.hpp"
#include "rpc_topic.hpp"

namespace RPC
{
    namespace client
    {
        class RegistryClient
        {
        public:
            using ptr = std::shared_ptr<RegistryClient>;
            // 传入注册中心的地址信息ip和port.
            RegistryClient(const std::string &ip, int port)
                : _reqeustor(std::make_shared<Requestor>()), _provider(std::make_shared<client::Provider>(_reqeustor)), _dispatcher(std::make_shared<RPC::Dispatcher>())
            {
                auto rsp_cb = std::bind(&RPC::client::Requestor::onResponse, _reqeustor.get(),
                                        std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<RPC::BaseMessage>(RPC::MType::RSP_SERVICE, rsp_cb);

                auto message_cb = std::bind(&RPC::Dispatcher::onMessage, _dispatcher.get(),
                                            std::placeholders::_1, std::placeholders::_2);
                _client = RPC::ClientFactory::create(ip, port);
                _client->setMessageCallback(message_cb);
                _client->connect();
            }
            // 向外提供服务注册接口;
            bool registryMethod(const std::string &method, const Address &host)
            {
                return _provider->registryMethod(_client->connection(), method, host);
            }

        private:
            Requestor::ptr _reqeustor;
            client::Provider::ptr _provider;
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _client;
        };

        class DiscoveryClient
        {
        public:
            using ptr = std::shared_ptr<DiscoveryClient>;

            DiscoveryClient(const std::string &ip, int port, const Discoverer::offlineCallback& cb)
                : _reqeustor(std::make_shared<Requestor>())
                , _discoverer(std::make_shared<client::Discoverer>(_reqeustor, cb))
                , _dispatcher(std::make_shared<RPC::Dispatcher>())
            {
                auto rsp_cb = std::bind(&RPC::client::Requestor::onResponse, _reqeustor.get(),
                                        std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(RPC::MType::RSP_SERVICE, rsp_cb);

                auto req_cb = std::bind(&client::Discoverer::onServiceRequest, _discoverer.get(),
                                        std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<ServiceRequest>(RPC::MType::REQ_SERVICE, req_cb);

                auto message_cb = std::bind(&RPC::Dispatcher::onMessage, _dispatcher.get(),
                                            std::placeholders::_1, std::placeholders::_2);
                _client = RPC::ClientFactory::create(ip, port);
                _client->setMessageCallback(message_cb);
                _client->connect();
            }
            // 服务发现接口;
            bool serviceDiscovery(const std::string &method, Address &host)
            {
                return _discoverer->serviceDiscovery(_client->connection(), method, host);
            }

        private:
            Requestor::ptr _reqeustor;
            client::Discoverer::ptr _discoverer;
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _client;
        };

        class RpcClient
        {
        public:
            using ptr = std::shared_ptr<RpcClient>;
            // 是否启用服务发现功能, 决定传入地址信息是服务提供者地址还是注册中心地址.
            RpcClient(bool enableDiscovery, const std::string &ip, int port)
                : _enableDiscovery(enableDiscovery)
                ,_reqeustor(std::make_shared<Requestor>())
                , _dispatcher(std::make_shared<RPC::Dispatcher>())
                , _caller(std::make_shared<RPC::client::RpcCaller>(_reqeustor))
            {
                // 针对rpc请求的响应进行回调处理;
                auto rsp_cb = std::bind(&RPC::client::Requestor::onResponse, _reqeustor.get(),
                                        std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(RPC::MType::RSP_RPC, rsp_cb);

                // 如果启用了服务发现, 地址信息是注册中心的地址, 是服务发现客户端连接的地址, 就通过地址信息实例化discovery_client;
                // 没有启用服务器就是服务提供者的地址信息, 直接实例化rpc_client;
                if (_enableDiscovery)
                {
                    auto offline_cb = std::bind(&RpcClient::delClient, this, std::placeholders::_1);
                    _discovery_client = std::make_shared<DiscoveryClient>(ip, port, offline_cb);
                }
                else
                {
                    auto message_cb = std::bind(&RPC::Dispatcher::onMessage, _dispatcher.get(),
                                                std::placeholders::_1, std::placeholders::_2);
                    _rpc_client = RPC::ClientFactory::create(ip, port);
                    _rpc_client->setMessageCallback(message_cb);
                    _rpc_client->connect();
                }
            }


            bool call(const std::string &method, const Json::Value &params, Json::Value &result)
            {
                // 获取服务提供者, 1.服务发现,  2. 固定服务提供者;
                BaseClient::ptr client = getClient(method);
                if(client.get() == nullptr)
                {
                        return false;
                }

                //通过客户端连接, 发起rpc请求;
                return _caller->call(client->connection(), method, params, result);
            }

            bool call(const std::string &method, const Json::Value &params, RpcCaller::JsonAsyncResponse &result)
            {
                BaseClient::ptr client = getClient(method);
                if(client.get() == nullptr)
                {
                        return false;
                }
                //通过客户端连接, 发起rpc请求;
                return _caller->call(client->connection(), method, params, result);
            }

            bool call(const std::string &method, const Json::Value &params, RpcCaller::JsonResponseCallback &cb)
            {
                 BaseClient::ptr client = getClient(method);
                if(client.get() == nullptr)
                {
                        return false;
                }
                //通过客户端连接, 发起rpc请求;
                return _caller->call(client->connection(), method, params, cb);
            }

        private:
            BaseClient::ptr newClient(const Address &host)
            {
                auto message_cb = std::bind(&RPC::Dispatcher::onMessage, _dispatcher.get(),
                    std::placeholders::_1, std::placeholders::_2);
                auto client = RPC::ClientFactory::create(host.first, host.second);
                client->setMessageCallback(message_cb);
                client->connect();
                putClient(host, client);
                return client;
            }

            BaseClient::ptr getClient(const Address &host)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _rpc_clients.find(host);
                if (it == _rpc_clients.end())
                {
                    return BaseClient::ptr();
                }
                return it->second;
            }

            BaseClient::ptr getClient(const std::string method)
            {
                BaseClient::ptr client;
                if (_enableDiscovery)
                {
                    // 1. 获取服务者地址信息;
                    Address host;
                    bool ret = _discovery_client->serviceDiscovery(method, host);
                    if (ret == false)
                    {
                        ELOG("当前%s服务, 没有找到服务提供者! ", method.c_str());
                        return  BaseClient::ptr();
                    }
                    // 2. 服务提供者是否已经实例化, 有就直接使用, 没有就创建;
                    client = getClient(host);
                    if (client.get() == nullptr)
                    {
                        // 没有找到实例化的客户端, 就创建;
                        client = newClient(host);
                    }
                }
                else
                {
                    client = _rpc_client;
                }
                return client;
            }

            void putClient(const Address &host, BaseClient::ptr &client)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _rpc_clients.insert(std::make_pair(host, client));
            }

            void delClient(const Address &host)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _rpc_clients.erase(host);
            }

        private:
            struct AddressHassh
            {
                size_t operator()(const Address &host)
                {
                    // ip+port;
                    std::string addr = host.first + std::to_string(host.second);
                    return std::hash<std::string>{}(addr);
                }
            };
            Requestor::ptr _reqeustor;
            bool _enableDiscovery;
            std::mutex _mutex;
            DiscoveryClient::ptr _discovery_client;
            RpcCaller::ptr _caller;
            BaseClient::ptr _rpc_client; // 未启用服务发现.
            Dispatcher::ptr _dispatcher;
            //<Add, [client1];
            std::unordered_map<Address, BaseClient::ptr, AddressHassh> _rpc_clients; // 服务发现的连接池;
        };


        class TopicClient
        {
        public:
            TopicClient(const std::string& ip, int port)
                :_reqeustor(std::make_shared<Requestor>())
                , _dispatcher(std::make_shared<RPC::Dispatcher>())
                , _topic_manager(std::make_shared<RPC::client::TopicManager>(_reqeustor))
            {
                 auto rsp_cb = std::bind(&RPC::client::Requestor::onResponse, _reqeustor.get(),
                        std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(RPC::MType::RSP_TOPIC, rsp_cb);

                auto msg_cb = std::bind(&TopicManager::publish, _topic_manager.get(),
                    std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<TopicRequest>(MType::REQ_TOPIC, msg_cb);


                auto message_cb = std::bind(&RPC::Dispatcher::onMessage, _dispatcher.get(),
                    std::placeholders::_1, std::placeholders::_2);
                _rpc_client = RPC::ClientFactory::create(ip, port);
                _rpc_client->setMessageCallback(message_cb);
                _rpc_client->connect();
            }

            bool create(const std::string &key)
            {
                return _topic_manager->create(_rpc_client->connection(), key);
            }

            void remove(const std::string &key)
            {
                return _topic_manager->remove(_rpc_client->connection(), key);
            }
            
            bool subscribe(const std::string &key, const TopicManager::SubCallback& cb)
            {
                return _topic_manager->subscribe(_rpc_client->connection(), key, cb);
              
            }

            bool cancel(const std::string& key)
            {
                return _topic_manager->cancel(_rpc_client->connection(), key);
            }

            bool publish(const std::string& key, const std::string& msg)
            {
                return _topic_manager->publish(_rpc_client->connection(), key, msg);
            }

            void shutdown()
            {
                _rpc_client->shutdown();
            }
        private:
            Requestor::ptr _reqeustor;
            TopicManager::ptr _topic_manager;
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _rpc_client;
        };
    }
}